package com.kafka.service;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * Kafka消息消费者服务
 * 提供以下功能：
 * 1. 监听并消费指定主题的消息
 * 2. 手动提交偏移量控制
 * 3. 错误消息处理机制
 * 4. 消息处理业务逻辑封装
 * 
 * 使用@KafkaListener注解配置监听主题和消费者组
 * 内置完善的日志记录和异常处理机制
 */
@Service
public class KafkaConsumerService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerService.class);

    /**
     * 监听并消费指定主题的消息
     * @param record ConsumerRecord对象，包含消息元数据(主题、分区、偏移量)和内容
     * @param ack Acknowledgment对象，用于手动提交偏移量
     * @throws RuntimeException 当消息处理失败时抛出
     * 
     * 消息处理流程：
     * 1. 记录接收到的消息日志
     * 2. 调用processMessage处理业务逻辑
     * 3. 手动提交偏移量(ack.acknowledge())
     * 
     * 异常处理策略：
     * - 记录错误日志
     * - 不提交偏移量，允许消息重试
     * - 可根据业务需求扩展为死信队列处理
     * 
     * 配置说明：
     * - 监听主题通过${spring.kafka.consumer.topic}配置，默认为default-topic
     * - 消费者组通过${spring.kafka.consumer.group-id}配置
     */
    @KafkaListener(topics = "${spring.kafka.consumer.topic:default-topic}", 
                  groupId = "${spring.kafka.consumer.group-id}")
    public void consumeMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            logger.info("Received message - Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}",
                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
            
            // 处理消息业务逻辑
            processMessage(record.value());
            
            // 手动提交偏移量
            ack.acknowledge();
            logger.info("Message processed and offset committed");
        } catch (Exception e) {
            logger.error("Error processing message", e);
            // 可根据业务需求决定是否重试或记录死信队列
        }
    }

    /**
     * 处理消息业务逻辑(内部方法)
     * @param message 消息内容(不能为空或空字符串)
     * @throws IllegalArgumentException 当消息为空或格式无效时抛出
     * 
     * 业务处理细节：
     * 1. 验证消息非空
     * 2. 记录处理日志
     * 3. 解析消息内容(JSON格式)
     * 4. 根据消息类型分发到具体处理器
     * 
     * 扩展建议：
     * - 可添加@Transactional注解支持事务
     * - 可引入消息处理策略模式
     * - 可添加重试机制处理临时故障
     * 
     * 性能考虑：
     * - 长时间处理应考虑异步化
     * - 批量处理可提高吞吐量
     */
    private void processMessage(String message) {
        // 实现具体的业务逻辑处理
        logger.info("Processing message: {}", message);
        
        // 示例：简单的消息处理
        if (message != null && !message.isEmpty()) {
            // 这里可以添加实际业务处理逻辑
            logger.info("Message processed successfully");
        }
    }

    /**
     * 监听并处理错误主题消息(死信队列)
     * @param record 错误消息记录，包含原始消息和错误信息
     * 
     * 错误处理策略：
     * 1. 记录错误详情到日志系统
     * 2. 持久化错误消息到数据库(可选)
     * 3. 触发告警通知(邮件/短信)
     * 
     * 配置说明：
     * - 错误主题通过${spring.kafka.consumer.error-topic}配置，默认为error-topic
     * - 建议为每个业务主题配置独立的错误主题
     * 
     * 扩展建议：
     * - 可添加错误消息重试机制
     * - 可集成第三方监控系统
     * - 可添加人工干预接口
     */
    @KafkaListener(topics = "${spring.kafka.consumer.error-topic:error-topic}")
    public void consumeErrorMessage(ConsumerRecord<String, String> record) {
        logger.error("Received error message - Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
        // 处理错误消息的逻辑
    }
}